package com.morpheus.zookeeper.api.guid;

import java.util.List;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.morpheus.zookeeper.api.client.ZookeeperClient;

/**
 * 
 */
public final class ZookeeperWorker {
	private final static Logger LOGGER = LoggerFactory.getLogger(ZookeeperWorker.class);

	private static final Long MAX_NODE_NUM = 1024L;
	private static final String PARENT_FOREVER_PATH = "/leaf-snowflake";
	private static final String APP_PATH = "/appName";
	private static final String PARENT_PATH = PARENT_FOREVER_PATH + APP_PATH;
	/**
	 * 临时节点
	 */
	private static final String TEMP_CHILD_PATH = "/instanceId";

	private static final String COMPLETE_PATH = PARENT_FOREVER_PATH + APP_PATH + TEMP_CHILD_PATH;

	private static ZookeeperClient zookeeperClient = null;

	private volatile static boolean initialized = false;

	private static volatile Long workId = null;

	public static Long getWorkId() {
		return workId;
	}

	public synchronized static void init(ZookeeperClient zookeeperClient) {
		LOGGER.debug("ZookeeperWorker.init() zookeeperClient={}", zookeeperClient);
		if (initialized) {
			return;
		}
		initialized = true;
		try {
			ZookeeperWorker.zookeeperClient = zookeeperClient;
			LOGGER.debug("ZookeeperWorker.init() ZookeeperWorker.zookeeperClient={}", ZookeeperWorker.zookeeperClient);
			createByZookeeper();
			addWatch(ZookeeperWorker.zookeeperClient);
		} catch (Throwable th) {
			LOGGER.error("ZookeeperWorker.init() throwable", th);
		}
	}

	private synchronized static void createByZookeeper() {
		List<String> children;
		try {
			children = zookeeperClient.getChildren(PARENT_PATH);
			if (children.isEmpty()) {
				zookeeperClient.delete(PARENT_PATH);
				zookeeperClient.createPersistent(PARENT_PATH, true);
			}
		} catch (Throwable th) {
			zookeeperClient.createPersistent(PARENT_PATH, true);
		}

		String data = String.valueOf(System.currentTimeMillis());

		// 创建临时顺序节点, 获取id
		String resultPath;
		try {
			resultPath = zookeeperClient.createEphemeralSequential(COMPLETE_PATH, data);
		} catch (Throwable th) {
			zookeeperClient.createPersistent(PARENT_PATH, true);
			resultPath = zookeeperClient.createEphemeralSequential(COMPLETE_PATH, data);
		}

		Long id = getIdFromPath(resultPath);
		if (id < MAX_NODE_NUM) {
			createWork(id);
			LOGGER.debug("ZookeeperWorker.createByZookeeper() id={}", id);
			return;
		}

		zookeeperClient.delete(resultPath);

		children = zookeeperClient.getChildren(PARENT_PATH);
		Set<Long> ids = children.stream().map(item -> getIdFromChildPath(item)).filter(item -> item != null)
				.collect(Collectors.toSet());
		LOGGER.debug("ZookeeperWorker.createByZookeeper() ids={}", ids);
		for (Long idx = 0L; idx < MAX_NODE_NUM; idx++) {
			if (!ids.contains(idx)) {
				try {
					String path = COMPLETE_PATH + String.format("%010d", idx);
					zookeeperClient.createEphemeral(path, data);
					createWork(id);
					LOGGER.debug("ZookeeperWorker.createByZookeeper() id={}", id);
					return;
				} catch (Exception e) {
					continue;
				}
			}
		}
		id = (id >= MAX_NODE_NUM) ? id % MAX_NODE_NUM : id;
		createWork(id);
	}

	private static void addWatch(ZookeeperClient zookeeperClient) throws Exception {
		Executor executor = Executors.newFixedThreadPool(2);

		ConnectionStateListener connectionStateListener = new ConnectionStateListener() {
			public void stateChanged(CuratorFramework client, ConnectionState newState) {
				if (newState == ConnectionState.CONNECTED) {
					LOGGER.debug("ZookeeperWorker.addWatch()  connected established");
					createByZookeeper();
				} else if (newState == ConnectionState.LOST) {
					LOGGER.debug("ZookeeperWorker.addWatch()  connection lost");
				}
			}
		};
		zookeeperClient.addConnectListener(connectionStateListener);

		PathChildrenCacheListener pathChildrenCacheListener = new PathChildrenCacheListener() {
			@Override
			public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
				switch (event.getType()) {
				case CHILD_ADDED:
					LOGGER.debug("Node added: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
					break;
				case CHILD_UPDATED:
					LOGGER.debug("Node changed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
					break;
				case CHILD_REMOVED:
					LOGGER.debug("Node removed: " + ZKPaths.getNodeFromPath(event.getData().getPath()));
					createByZookeeper();
					break;
				}
			}
		};
		zookeeperClient.addPathListener(PARENT_PATH, executor, pathChildrenCacheListener);
	}

	private static Long getIdFromChildPath(String path) {
		try {
			String idStr = path.replace("instanceId", "");
			return Long.valueOf(idStr);
		} catch (Throwable th) {
			LOGGER.error("ZookeeperWorker.getIdFromChildPath() throwable", th);
		}
		return 0L;
	}

	private static Long getIdFromPath(String path) {
		String idStr = path.replace(COMPLETE_PATH, "");
		return Long.valueOf(idStr);
	}

	private static void createWork(Long id) {
		workId = id;
		SnowFlakeWorker snowFlakeWorker = SnowFlakeWorker.initSnowFlakeWorker(id);
		LOGGER.debug("ZookeeperWorker.createWork() snowFlakeWorker={}", snowFlakeWorker);
	}
}
